/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.example.benchmark;

import java.util.LinkedList;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;


/**
 * 性能测试，多线程同步发送消息
 */
public class DelayProducer {
	private static int limit = 0;
	private static int delayLevel = 0;
	private static int threadCount;
	private static int messageSize;
	private static int sleepTime = 0;

	private static String topic;

	private static AtomicLong totalSend = new AtomicLong();

    public static void main(String[] args) throws MQClientException {
		delayLevel = args.length >= 1 ? Integer.parseInt(args[0]) : 2;
		threadCount = args.length >= 2 ? Integer.parseInt(args[1]) : 1;
		messageSize = args.length >= 3 ? Integer.parseInt(args[2]) : 256;
		topic = args.length >= 4 ? args[3] : "BenchmarkTest";
		sleepTime = args.length >= 5 ? Integer.parseInt(args[4]) : 0;
		limit = args.length >= 6 ? Integer.parseInt(args[5]) : 100000;

        System.out.printf("threadCount %d messageSize %d\n", threadCount, messageSize);

		/**
		 * 超过4K压缩，模拟一半压缩，一半不压缩的方式
		 */
		final byte[] body1 = buildBody(messageSize);
		final byte[] body2;
		if (messageSize > 400) {
			body2 = buildBody(300);
		} else {
			body2 = body1;
		}

        final ExecutorService sendThreadPool = Executors.newFixedThreadPool(threadCount);
        final StatsBenchmarkProducer statsBenchmark = new StatsBenchmarkProducer();
        final Timer timer = new Timer("BenchmarkTimerThread", true);
        final LinkedList<Long[]> snapshotList = new LinkedList<Long[]>();

        timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                snapshotList.addLast(statsBenchmark.createSnapshot());
                if (snapshotList.size() > 10) {
                    snapshotList.removeFirst();
                }
            }
        }, 1000, 1000);

        timer.scheduleAtFixedRate(new TimerTask() {
            private void printStats() {
                if (snapshotList.size() >= 10) {
                    Long[] begin = snapshotList.getFirst();
                    Long[] end = snapshotList.getLast();

                    final long sendTps =
                            (long) (((end[3] - begin[3]) / (double) (end[0] - begin[0])) * 1000L);
                    final double averageRT = ((end[5] - begin[5]) / (double) (end[3] - begin[3]));

                    System.out.printf(
                        "Send TPS: %d Max RT: %d Average RT: %7.3f Send Failed: %d Response Failed: %d\n"//
                        , sendTps//
                        , statsBenchmark.getSendMessageMaxRT().get()//
                        , averageRT//
                        , end[2]//
                        , end[4]//
                        );
					System.out.println("totalSend:" + totalSend.get());
                }
            }


            @Override
            public void run() {
                try {
                    this.printStats();
                }
                catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, 10000, 10000);

        final DefaultMQProducer producer = new DefaultMQProducer("benchmark_producer");
        producer.setInstanceName(Long.toString(System.currentTimeMillis()));
        producer.setCompressMsgBodyOverHowmuch(Integer.MAX_VALUE);
		//producer.setNamesrvAddr("127.0.0.1:9876");

        producer.start();

        for (int i = 0; i < threadCount; i++) {
            sendThreadPool.execute(new Runnable() {
                @Override
                public void run() {
					int i = 0;
                    while (true) {
                        try {
                            final long beginTimestamp = System.currentTimeMillis();
							Message msg = null;
							if (i >= Integer.MAX_VALUE) {
								i = 0;
							}
							if (i % 2 == 0) {
								msg = buildMessage(body1);
							} else {
								msg = buildMessage(body2);
							}
							i++;
							/**
							 * send至多重试3次,超时不会重试
							 */
							msg.setKeys(totalSend.incrementAndGet() + "");
							producer.send(msg);

                            statsBenchmark.getSendRequestSuccessCount().incrementAndGet();
                            statsBenchmark.getReceiveResponseSuccessCount().incrementAndGet();
                            final long currentRT = System.currentTimeMillis() - beginTimestamp;
                            statsBenchmark.getSendMessageSuccessTimeTotal().addAndGet(currentRT);
                            long prevMaxRT = statsBenchmark.getSendMessageMaxRT().get();
                            while (currentRT > prevMaxRT) {
                                boolean updated =
                                        statsBenchmark.getSendMessageMaxRT().compareAndSet(prevMaxRT,
                                            currentRT);
                                if (updated)
                                    break;

                                prevMaxRT = statsBenchmark.getSendMessageMaxRT().get();
                            }
							if (i == limit) {
								break;
							}
							if (sleepTime > 0) {
								Thread.sleep(sleepTime);
							}
                        }
                        catch (RemotingException e) {
                            statsBenchmark.getSendRequestFailedCount().incrementAndGet();
                            e.printStackTrace();
                        }
                        catch (InterruptedException e) {
                            statsBenchmark.getSendRequestFailedCount().incrementAndGet();
                            e.printStackTrace();
                        }

                        catch (MQClientException e) {
                            statsBenchmark.getSendRequestFailedCount().incrementAndGet();
                            e.printStackTrace();
                        }
                        catch (MQBrokerException e) {
                            statsBenchmark.getReceiveResponseFailedCount().incrementAndGet();
                            e.printStackTrace();
                        }
                    }
					System.out.println("send end:" + i);
                }
            });
        }
    }

	private static byte[] buildBody(final int messageSize) {
		StringBuilder sb = new StringBuilder();
		for (int i = 0; i < messageSize; i += 10) {
			sb.append("hello baby");
		}
		// 记录下当前时间
		// sb.append("|").append(System.currentTimeMillis());
		return sb.toString().getBytes();
	}

	private static Message buildMessage(final byte[] body) {
        Message msg = new Message();
		msg.setTopic(topic);
		msg.setDelayTimeLevel(delayLevel);
		msg.setBody(body);
        return msg;
    }
}
